package com.atguigu.gmall.realtime.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.util.DruidDSUtil;
import com.atguigu.gmall.realtime.util.PhoenixUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * Author: Felix
 * Date: 2022/6/11
 * Desc: 从主流中过滤出维度数据
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private DruidDataSource dataSource;

    public TableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //获取连接池对象
        dataSource = DruidDSUtil.createDataSource();
    }

    //{"database":"gmall1216","xid":22634,"data":{"tm_name":"大米","logo_url":"/static/default.jpg","id":5},
    // "old":{"tm_name":"小米"},"commit":true,"type":"update","table":"base_trademark","ts":1654928829}
    //处理主流业务数据
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //获取当前处理的业务数据库的表名
        String tableName = jsonObj.getString("table");
        //根据业务数据库表名到广播状态中获取对应的配置信息
        TableProcess tableProcess = broadcastState.get(tableName);
        if(tableProcess != null){
            //说明根据当前处理的业务数据库表名 到广播状态中找到了对应的配置信息，说明当前处理的是维度数据，将其向下游传递
            //获取维度表中一条数据    {"tm_name":"大米","logo_url":"/static/default.jpg","id":5}
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");

            //在向下游传递数据前，将不需要传递的属性过滤掉   {"tm_name":"大米","id":5}
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumn(dataJsonObj,sinkColumns);

            //在向下游传递数据前，应该补充保存的目的地  {"tm_name":"大米","id":5,"sink_table":"dim_base_trademark"}
            String sinkTable = tableProcess.getSinkTable();
            dataJsonObj.put("sink_table",sinkTable);

            //在向下游传递数据前，应该补充操作类型
            String type = jsonObj.getString("type");
            dataJsonObj.put("type",type);

            //将当前处理的维度数据 传递到下游
            out.collect(dataJsonObj);
        }else{
            //说明根据当前处理的业务数据库表名 到广播状态中没有找到对应的配置信息，说明当前处理的不是维度数据，不对其进行处理，相当于过滤掉了
            System.out.println("不是维度，过滤掉:" + jsonObj);
        }

    }

    /*
     {
        "before":{"source_table":"t_test","sink_table":"dim_t_test1111","sink_columns":"id,name","sink_pk":"id","sink_extend":null},
        "after":{"source_table":"t_test","sink_table":"dim_t_test2222","sink_columns":"id,name","sink_pk":"id","sink_extend":null},
        "source":{"version":"1.5.4.Final","connector":"mysql","db":"gmall1216_config","sequence":null,"table":"table_process",
            "server_id":1,"gtid":null,"file":"mysql-bin.000013","pos":1476,"row":0,"thread":null,"query":null},
        "op":"u",
        "ts_ms":1654910481664,
        "transaction":null
        }
    */
    //处理广播流配置数据
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        //为了处理方便  将流中的json字符串转换为json对象
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        //获取广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //获取对配置表的操作类型
        String op = jsonObj.getString("op");

        if ("d".equals(op)) {
            //如果是删除操作   将状态中配置删除掉
            TableProcess before = jsonObj.getObject("before", TableProcess.class);
            String sourceTable = before.getSourceTable();
            broadcastState.remove(sourceTable);
        } else {
            //如果是删除之外的其它操作  r:read 读  c:create 新增   u:update 更新   将对应的配置 放到广播状态中
            //获取配置表中的一条配置记录  并封装为实体类对象
            TableProcess tableProcess = jsonObj.getObject("after", TableProcess.class);
            //获取业务数据库表名
            String sourceTable = tableProcess.getSourceTable();
            //获取数仓中维度表名
            String sinkTable = tableProcess.getSinkTable();
            //获取维度表中字段        1.建表时候用       2.过滤字段用
            String sinkColumns = tableProcess.getSinkColumns();
            //获取维度表主键
            String sinkPk = tableProcess.getSinkPk();
            //获取建表扩展
            String sinkExtend = tableProcess.getSinkExtend();

            //提前将phoenix中的维度表创建出来
            checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);

            //将配置表中的配置放到广播状态中
            broadcastState.put(sourceTable, tableProcess);
        }
    }

    //过滤不需要传递的属性    dataJsonObj: {"tm_name":"大米","logo_url":"/static/default.jpg","id":5}   sinkColumns: id,tm_name
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);

        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        entrySet.removeIf(entry->!columnList.contains(entry.getKey()));
    }

    //创建phoenix中的维度表
    private void checkTable(String tableName, String columns, String pk, String ext) {
        if (pk == null) {
            pk = "id";
        }

        if (ext == null) {
            ext = "";
        }
        //拼接建表语句
        StringBuilder createSql = new StringBuilder("create table if not exists " + GmallConfig.PHOENIX_SCHEMA + "." + tableName + "(");
        String[] columnArr = columns.split(",");
        for (int i = 0; i < columnArr.length; i++) {
            String column = columnArr[i];
            if (column.equals(pk)) {
                createSql.append(column + " varchar primary key ");
            } else {
                createSql.append(column + " varchar ");
            }
            if (i < columnArr.length - 1) {
                createSql.append(",");
            }
        }
        createSql.append(") " + ext);
        System.out.println("在phoenix表中建表语句为: " + createSql);

        try {
            Connection conn  = dataSource.getConnection();
            PhoenixUtil.executeSql(createSql.toString(),conn);
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}